package com.om.opensourway;

import Utils.HttpClientUtils;
import com.om.common.CommonObject;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.codehaus.jackson.JsonNode;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;

import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.UUID;

/*
 * 根据vhost，进行数据过滤，将过滤后的数据按照特定的格式{\"country\":\"" + country + "\",\"path\":\"" + path + "\",\"hostname\":\"" + hostname + "\",\"updated_at\":\"" + t + "\",\"city\":\"" + city + "\",\"ip\":\"" + ip + "\",\"link\":\"" + link + "\",\"created_at\":\"" + t + "\",\"location_ip\":\"" + locationIp + "\",\"location\":{\"lon\":" + lon + ",\"lat\":" + lat + "}}"
 * 存放到es 的对应传入的index 中。
 * 命令：/opt/client/Spark/spark/bin/spark-submit --class com.om.opensourway.AnalysisForBatch --num-executors 1 --executor-cores 6 --master yarn-cluster /root/om-bigdata-1.0-SNAPSHOT.jar ak sk 6 inputdatafile vhostname indexname appname
 * */
public class AnalysisForBatch {

    public static void main(String[] args) throws URISyntaxException, ClassNotFoundException, IllegalAccessException, InstantiationException {
        String taskCount = args[2];
        String inputPath = args[3];
        String[] VHOSTS = args[4].split(",");
        String esIndexName = args[5];
        String className = args[7];
        CommonObject commonObject = (CommonObject) Class.forName(className).newInstance();
        System.out.println("参数args：" + Arrays.asList(args).toString());
        commonObject.setEsIndexName(esIndexName);
        commonObject.setVHOSTS(VHOSTS);
        SparkConf sparkConf = commonObject.getSparkConf(args);
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        Broadcast<CommonObject> broCommonObject = jsc.broadcast(commonObject);
        JavaRDD<String> srcRdd = jsc.textFile(inputPath, Integer.parseInt(taskCount));
        System.out.println("srcrdd's partitions: " + srcRdd.getNumPartitions());
        switch (className) {
            case "com.om.Module.openeuler.logstash_repo_openeuler_org":
                JavaRDD filterDatas = broCommonObject.value().dealData(srcRdd, broCommonObject.value().getVHOSTS());
                filterDatas.foreachPartition(new VoidFunction<Iterator<JsonNode>>() {
                    final StringBuffer resultBuffer = new StringBuffer();

                    @Override
                    public void call(Iterator<JsonNode> jsonNodeIterator) throws Exception {
                        CloseableHttpClient client = HttpClientUtils.getClient();
                        RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
                        BulkRequest request = new BulkRequest();

                        while (jsonNodeIterator.hasNext()) {
                            JsonNode record = jsonNodeIterator.next();
                            JsonNode sourceNode = record.get("_source");
                            String logstr = sourceNode.get("log").getTextValue();
                            JsonNode log = CommonObject.objectMapper.readTree(logstr);
                            HashMap<String, String> resMap = broCommonObject.value().assemblydataNew(client, log, record.get("_id").getTextValue());
                            String hostname = resMap.get("hostname");
                            String id = ("repo.openeuler.org".equals(hostname) || "download.openlookeng.io".equals(hostname)) ?
                                    DigestUtils.md5Hex(resMap.get("created_at") + resMap.get("ip") + resMap.get("path")) :
                                    record.get("_id").getTextValue();
                            //如果任务提交参数中esindexname为”-“数据存入11.4 否则数据存入指定的esindexname
                                if(broCommonObject.value().getEsIndexName().equals("-")){
                                    request.add(new IndexRequest(hostname + "_20201104", "_doc", id).source(resMap));
                                }else {
                                    request.add(new IndexRequest(broCommonObject.value().getEsIndexName(), "_doc", id).source(resMap));
                                }
                        }
                        if (request.requests().size() != 0)
                            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
                        restHighLevelClient.close();
                    }
                });
                long foreachPartitionstart = System.currentTimeMillis();
                long foreachpartitionsend = System.currentTimeMillis();
                System.out.println("foreachpartation spend: " + (foreachpartitionsend - foreachPartitionstart) + " milliseconds");
                break;
            case "com.om.Module.openeuler.morrirs_Openeuler_Recoeds":
                JavaRDD data = broCommonObject.value().dealData(srcRdd, broCommonObject.value().getVHOSTS());
                data.foreachPartition(new VoidFunction<Iterator>() {
                    final StringBuffer resultBuffer = new StringBuffer();
                    @Override
                    public void call(Iterator stringIterator) throws Exception {
                        while (stringIterator.hasNext()) {
                            String record = stringIterator.next().toString();
                            String assemblydata = "";
                            boolean isright = true;
                            String id="";
                            try {
                                HashMap resMap = broCommonObject.value().assemblydata(record, "");
                                assemblydata = CommonObject.objectMapper.writeValueAsString(resMap);
                                id=resMap.get("created_at").toString()+resMap.get("ip");

                            } catch (Exception e) {
                                System.err.println(e.getMessage() + record);
                                isright = false;
                            }
                            if (isright) {
                                String indexstr = "{\"index\":{\"_index\":\"" + broCommonObject.value().getEsIndexName() + "\",\"_id\":\"" + id + "\"}}";
                                resultBuffer.append(indexstr).append("\n").append(assemblydata).append("\n");
                            }
                        }
                        if (StringUtils.isNotBlank(resultBuffer.toString())) {
                            long start = System.currentTimeMillis();
                            int code = broCommonObject.value().sendData(resultBuffer.toString());
                            System.out.println(code);
                            System.out.println("send:" + resultBuffer.toString());
                            System.out.println("发送buffer数据（milliseconds）：" + (System.currentTimeMillis() - start) + " milliseconds");
                        }
                    }
                });
                break;

        }


    }
}


